from absl import app, flags
import datetime
import time

from coin.base.datetime_util import to_timestamp_int
from coin.base.param_util import split_string
import coin.base.database.util as db_util
import coin.support.accounting.logic.account_status_util as account_status
from xunkemgmt_client.support.accounting.database.mysql_importer import \
    MysqlAccountInfoImporter as MysqlImporter
from coin.support.accounting.logic.kafka_querier import KafkaAccountInfoQuerier
from coin.util.queue.config import KafkaConfig
from coin.support.accounting.app.account_info_collector import _load_proxy_config
from coin.support.accounting.logic.query_util import filter_invalid_request_and_result
from xunkemgmt_client.client.util.query_util import query_accounts

FLAGS = flags.FLAGS

EXCLUDE_MARKET_TYPE = ['MARGIN']

CURRENCY_LIST = ['ADA', 'BTC', 'EOS', 'ETH', 'LTC', 'TRX', 'XRP', 'USDT']


def _filter_transfer_by_days(transfer_responses, days=1):
  if days is not None:
    td_threshold = (datetime.datetime.utcnow() - datetime.timedelta(days=days)).date()
    ts_threshold = to_timestamp_int(td_threshold)
    for each_response in transfer_responses:
      transfers = each_response.transfer.each_transfer
      transfers.sort(key=lambda x: x.create_timestamp, reverse=True)
      for each_transfer in reversed(transfers):
        if each_transfer.create_timestamp < ts_threshold:
          transfers.pop()
        else:
          break
      transfers.sort(key=lambda x: x.create_timestamp)
  return transfer_responses


def py_collect_transfers(transfer_request, currency_list, proxy=None):
  for currency in currency_list:
    transfer_response = account_status.query_transfer_list(transfer_request, currency, proxy)
    transfer_request, transfer_response = \
        filter_invalid_request_and_result(transfer_request, transfer_response)
    transfer_response = _filter_transfer_by_days(transfer_response, FLAGS.days)
    db_config = db_util.read_db_config(FLAGS.mysql_config)
    importer = MysqlImporter(db_config=db_config)
    importer.upsert_transfer_history(transfer_request, transfer_response)


def cc_collect_transfers(transfer_request):
  transfer_request = [request for request in transfer_request 
                      if request.market_type.upper() not in EXCLUDE_MARKET_TYPE]
  kafka_config = KafkaConfig.from_cmd_config(FLAGS.kafka_config_filename)
  kafka_querier = KafkaAccountInfoQuerier(kafka_config)
  transfer_response = kafka_querier.query_transfer_history(transfer_request)
  transfer_request, transfer_response = \
        filter_invalid_request_and_result(transfer_request, transfer_response)
  db_config = db_util.read_db_config(FLAGS.mysql_config)
  importer = MysqlImporter(db_config=db_config)
  importer.upsert_transfer_history(transfer_request, transfer_response)


def main(_):
  start_time = time.time()
  request_list = query_accounts(active=True,
                                business_units=['Coin'],
                                exchanges=split_string(FLAGS.exchange),
                                market_types=split_string(FLAGS.market_type),
                                owners=split_string(FLAGS.owner),
                                as_proto=True)
  if len(request_list) == 0:
    return
  proxy = _load_proxy_config(FLAGS.proxy_config_file)
  currency_list = CURRENCY_LIST
  while True:
    if FLAGS.legacy:
      py_collect_transfers(request_list, currency_list, proxy)
    else:
      cc_collect_transfers(request_list)
    if FLAGS.duration is not None and \
       time.time() - start_time > FLAGS.duration:
      break
    time.sleep(FLAGS.query_frequency)


if __name__ == '__main__':
  flags.DEFINE_string('market_type', None, 'Comma separated market_type.')

  flags.DEFINE_string('exchange', None, 'Comma separated exchange.')

  flags.DEFINE_string('owner', None, 'Comma separated account owner.')

  flags.DEFINE_string('proxy_config_file', None, 'proxy_config_file')

  flags.DEFINE_integer('query_frequency', 30, 'query frequency in seconds.')

  flags.DEFINE_integer(
      'duration',
      None,
      'Duration for running collector in seconds. Eg. If set to 300, program will '
      'stop after running 300 seconds. If set to None, program will run forever')
  
  flags.DEFINE_string('kafka_config_filename', None, 'kafka config')

  flags.DEFINE_string('mysql_config',
                      "../../coin_deploy/support_accounting/db_config/mysql_config_prod.json",
                      'path of database config.')

  flags.DEFINE_integer('days', None, 'transfer history in last N days.')

  flags.DEFINE_boolean('legacy', False, 'py or cc transfer collector')

  app.run(main)
